package com.guli.product.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.guli.common.constant.EsConstant;
import com.guli.common.constant.MqConstant;
import com.guli.common.to.esmodel.AttrModel;
import com.guli.common.to.esmodel.SkuEsModel;
import com.guli.product.entity.SkuCanal;
import com.guli.product.vo.CanalMessage;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.List;


/**
 * @Author zzb
 * @Date 2024/8/13 23:14
 * @Version 1.0
 * @description
 */
@Slf4j
@Component
public class CanalListener {
    @Autowired
    private RestHighLevelClient esClient;

    // 搭建es+canal
    @RabbitListener(queues = MqConstant.PMS_CANAL_QUEUE)
    public void handleCanalMessage(Message message) {
        byte[] body = message.getBody();
        String msg = new String(body, StandardCharsets.UTF_8);
        CanalMessage canalMessage = JSONObject.parseObject(msg, CanalMessage.class);
        //判断SQL的类型
        String type = canalMessage.getType();
        List<SkuCanal> canalMessageData = canalMessage.getData();
        if ("UPDATE".equals(type)) {
            //更新操作
            canalMessageData.forEach(canalMsg -> {
                SkuEsModel skuEsModel = new SkuEsModel();
                BeanUtils.copyProperties(canalMsg, skuEsModel);
                List<AttrModel> attrModels = JSONObject.parseObject(canalMsg.getAttrs(), new TypeReference<List<AttrModel>>() {
                });
                skuEsModel.setAttrs(attrModels);
                UpdateRequest updateRequest = new UpdateRequest(EsConstant.PRODUCT_INDEX,
                        canalMsg.getSkuId().toString());
                updateRequest.doc(JSON.toJSONString(skuEsModel), XContentType.JSON);
                try {
                    esClient.update(updateRequest, RequestOptions.DEFAULT);
                } catch (IOException e) {
                    log.info(e.getMessage());
                }
            });
        } else if ("DELETE".equals(type)) {
            //删除操作
            canalMessageData.forEach(canalMsg -> {
                DeleteRequest deleteRequest = new DeleteRequest(EsConstant.PRODUCT_INDEX,
                        canalMsg.getSkuId().toString());
                try {
                    esClient.delete(deleteRequest, RequestOptions.DEFAULT);
                } catch (IOException e) {
                    log.info(e.getMessage());
                }
            });
        }
    }
}
